package my.thesis.bolts.persistorsReliable;

import java.sql.Connection;
import java.util.ArrayList;
import java.util.Map;

import my.thesis.analytics.metrics.LastTimePersistMetric;

import org.joda.time.DateTime;

import redis.clients.jedis.Jedis;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;

public class HashStreamPersistor2Reliable extends BaseRichBolt
{
	private static final long serialVersionUID = 1L;
	private Jedis jedis;
	private OutputCollector collector;
	Connection connect=null;
	Integer timeIntervalSec;
	
	transient LastTimePersistMetric _lastTimeMetric;
	private static Boolean metricsOn=false;

	private String database;
	
	public void prepare(Map stormConf, TopologyContext context,OutputCollector collector) 
	{
		jedis=new Jedis(this.database);
		jedis.select(8);
		this.collector=collector;
		
		if(metricsOn)
		{
			_lastTimeMetric=new LastTimePersistMetric();
			context.registerMetric("sameHashtagPersist", _lastTimeMetric, 50);
			_lastTimeMetric.setStartTime();
		}
	}
		
	public void declareOutputFields(OutputFieldsDeclarer declarer) 
	{
		// TODO Auto-generated method stub
	}
	
	public HashStreamPersistor2Reliable(Integer ti,String connectionString )
	{
		this.timeIntervalSec=ti;
		this.database=connectionString;
	}
	
	public void execute(Tuple tuple) 
	{
		if(metricsOn)
		{
			_lastTimeMetric.setTimeReceived();
		}
		
		 Long timeInterval=tuple.getLongByField("timeInterval");
		 
		 Long timeExactMillis=timeInterval * timeIntervalSec * 1000;
		 
		 DateTime timeExact=new DateTime(timeExactMillis);
		
		 String timeInterval_s = timeExact.toString();

		 @SuppressWarnings("unchecked")
		 ArrayList<String> sameHashesForInterval = (ArrayList<String>) tuple.getValueByField("countsForInterval");
			
		 //TEST
		 System.out.println("Persisting same hashtags...");		

		 for(String h : sameHashesForInterval)	
		 {	 
			 try
			 {	 
		     	jedis.lpush(timeInterval_s, h);
		     	
			 }
			 catch(Exception ex)
			 {
				System.out.println("Error saving to redis");
				collector.fail(tuple);
				return;
			 }
		 }
		 
		 collector.ack(tuple);
	}
				
	public void cleanup()
	{
	    if (jedis.isConnected()) 
	    {
		      jedis.quit();
		}	
	}

}
